#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/msg.h>
#include <MQTTClient.h>
#include "cJSON.h"
#define ADDRESS        "tcp://192.168.247.228:1883"  // MQTT服务器地址
#define CLIENTID       "ExampleClientSub"
#define TOPIC          "/app/data/down"            // 订阅主题
#define RESPONSE_TOPIC "/app/data/up"          // 推送主题
#define QOS            1
#define TIMEOUT        10000L

int mode=1;

// 消息队列结构体
struct msg_buffer {
    long msg_type;
    char msg_text[100];
};

// 共享内存结构体
struct shared_memory {
    char data[100];
};

volatile MQTTClient_deliveryToken deliveredtoken;

// 消息发送确认回调
void delivered(void *context, MQTTClient_deliveryToken dt) {
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

// MQTT订阅消息回调
int msgarrvd(void* context, char* topicName, int topicLen, MQTTClient_message* message) {
	printf("Message arrived\n");
	printf("     topic: %s\n", topicName);
	printf("   message: %.*s\n", message->payloadlen, (char*)message->payload);
	cJSON* root = cJSON_Parse(message->payload);
	if (!root) {
		const char* error_ptr = cJSON_GetErrorPtr();
		if (error_ptr != NULL) {
			printf("JSON解析错误: %s\n", error_ptr);
		}
		else {
			printf("JSON解析错误\n");
		   // continue;
		}

	}
	// 提取 JSON 字段
    cJSON* mode_value = cJSON_GetObjectItem(root, "mode");
	if (mode_value != NULL) {
		mode = mode_value->valueint;
		printf("Mode changed to %d: Compare reporting\n", mode);
	}

	// 将接收到的消息放入消息队列
	int msgid = msgget(0x200, 0666 | IPC_CREAT);
	struct msg_buffer buffer;
	buffer.msg_type = 1;
	strncpy(buffer.msg_text, (char*)message->payload, message->payloadlen);
	buffer.msg_text[message->payloadlen] = '\0';
	if (msgsnd(msgid, &buffer, sizeof(buffer.msg_text), 0) == -1) {
		perror("Failed to send message to queue");
	}
	else {
		printf("Message send to message queue\n");
	}

	MQTTClient_freeMessage(&message);
	MQTTClient_free(topicName);
	return 1;
}

// 从消息队列读取数据并推送到MQTT
/*void *readFromMessageQueue(void *client) {
    MQTTClient mqttClient = (MQTTClient)client;
    int msgid = msgget(0x001, 0666 | IPC_CREAT);
    struct msg_buffer buffer;

    while (1) {
        if (msgrcv(msgid, &buffer, sizeof(buffer.msg_text), 1, 0) != -1) {
            printf("Message from queue: %s\n", buffer.msg_text);
            // 推送到MQTT服务器（推送主题）
            MQTTClient_publish(mqttClient, RESPONSE_TOPIC, strlen(buffer.msg_text), buffer.msg_text, QOS, 0, NULL);
            printf("Message sent to MQTT topic: %s\n", RESPONSE_TOPIC);
        }
        usleep(100000);  // 100ms
    }
}*/

// 从共享内存读取数据并推送到MQTT
void *readFromSharedMemory(void *client) {
    MQTTClient mqttClient = (MQTTClient)client;
    int shmid = shmget(0x300, sizeof(struct shared_memory), 0666 | IPC_CREAT);
    struct shared_memory * shm = shmat(shmid, NULL, 0);
    char previous_data[100] = "";  // 保存上次的数据用于对比

    while (1) {
        if (mode == 1) {  // 循环上报
            if (shm->data[0] != '\0') {
                printf("Message from shared memory (loop): %s\n", shm->data);
                MQTTClient_publish(mqttClient, RESPONSE_TOPIC, strlen(shm->data), shm->data, QOS, 0, NULL);
                printf("Message sent to MQTT topic: %s\n", RESPONSE_TOPIC);
            }
            usleep(5000000);  // 每5秒循环上报
        } else if (mode == 3) {  // 刷新上报
            if (shm->data[0] != '\0') {
                printf("Message from shared memory (refresh): %s\n", shm->data);
                MQTTClient_publish(mqttClient, RESPONSE_TOPIC, strlen(shm->data), shm->data, QOS, 0, NULL);
                printf("Message sent to MQTT topic: %s\n", RESPONSE_TOPIC);
                while(mode==3);
            }
            //mode = 1;
            usleep(100000);  // 100ms检查
        } else if (mode == 2) {  // 对比上报
            if (shm->data[0] != '\0' && strcmp(shm->data, previous_data) != 0) {
                printf("Message from shared memory (compare): %s\n", shm->data);
                MQTTClient_publish(mqttClient, RESPONSE_TOPIC, strlen(shm->data), shm->data, QOS, 0, NULL);
                printf("Message sent to MQTT topic: %s\n", RESPONSE_TOPIC);
                strncpy(previous_data, shm->data, sizeof(previous_data));  // 更新上次数据
            }
            usleep(100000);  // 100ms检查
        }
    }
}
// MQTT连接丢失回调
void connlost(void *context, char *cause) {
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

int main(int argc, char *argv[]) {
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    int ch;

    // 初始化MQTT客户端
    MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    // 连接到MQTT服务器
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    printf("Subscribing to topic %s\nfor client %s using QoS%d\n\n"
           "Press Q<Enter> to quit\n\n", TOPIC, CLIENTID, QOS);
    MQTTClient_subscribe(client, TOPIC, QOS);  // 订阅主题

    pthread_t thread1, thread2;
    //pthread_create(&thread1, NULL, readFromMessageQueue, (void *)client);
    pthread_create(&thread2, NULL, readFromSharedMemory, (void *)client);

    do {
        ch = getchar();
    } while (ch != 'Q' && ch != 'q');

    MQTTClient_unsubscribe(client, TOPIC);  // 取消订阅
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);

    // 删除消息队列
    int msgid = msgget(0x001, 0666 | IPC_CREAT);
    msgctl(msgid, IPC_RMID, NULL);

    // 删除共享内存
    int shmid = shmget(0x124, sizeof(struct shared_memory), 0666 | IPC_CREAT);
    shmctl(shmid, IPC_RMID, NULL);

    return rc;
}
